package com.umeox.moto.mqtt;

import android.text.TextUtils;
import android.util.Log;


import com.umeox.moto.mqtt.base.AError;
import com.umeox.moto.mqtt.base.ARequest;
import com.umeox.moto.mqtt.base.AResponse;
import com.umeox.moto.mqtt.base.IOnCallListener;
import com.umeox.moto.mqtt.base.PersistentConnectState;
import com.umeox.moto.mqtt.event.IOnRrpcResponseHandle;
import com.umeox.moto.mqtt.event.IOnSubscribeRpcListener;
import com.umeox.moto.mqtt.event.PersistentEventDispatcher;
import com.umeox.moto.mqtt.request.MqttPublishRequest;
import com.umeox.moto.mqtt.request.MqttRrpcRequest;

import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttAsyncClient;
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;

import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;

public class MqttDefaultCallback implements MqttCallbackExtended {

    private Map<String, IOnSubscribeRpcListener> listenerMap;

    private int connectionLostReasonCode;

    public MqttDefaultCallback() {

    }

    public void registerRrpcListener(String topic, IOnSubscribeRpcListener listener) {
        if (BuildConfig.DEBUG) {
            Log.d("MqttDefaultCallback", "registerRrpcListener(), topic = " + topic);
        }
        if (!TextUtils.isEmpty(topic) && listener != null) {
            if (listenerMap == null) {
                listenerMap = new HashMap<>();
            }
            listenerMap.put(topic, listener);
        } else {
            if (BuildConfig.DEBUG) {
                Log.d("MqttDefaultCallback", "registerRrpcListener(), params error ");
            }
        }
    }

    public void unRegisterRrpcListener(String topic) {
        if (BuildConfig.DEBUG) {
            Log.d("MqttDefaultCallback", "unRegisterRrpcListener(), topic = " + topic);
        }
        if (!TextUtils.isEmpty(topic)) {
            if (listenerMap == null) {
                return;
            }
            listenerMap.remove(topic);
        } else {
            if (BuildConfig.DEBUG) {
                Log.d("MqttDefaultCallback", "unRegisterRrpcListener(), params error ");
            }
        }
    }

    @Override
    public void connectComplete(boolean reconnect, String serverURI) {
        if (BuildConfig.DEBUG) {
            Log.d("MqttDefaultCallback", "connectComplete,reconnect = " + reconnect + " ," + serverURI);
        }
        if (!reconnect) {
            MqttNet.getInstance().changeConnectState(PersistentConnectState.CONNECTED);
            PersistentEventDispatcher.getInstance().handlerMessage(PersistentEventDispatcher.MSG_CONNECT_STATE_CONNECTED);
        } else {
            // 目前测试客户端断网或者服务器关闭都是返回REASON_CODE_CONNECTION_LOST（32109）
//            int reconnectTime = 1;
//            if (connectionLostReasonCode == MqttException.REASON_CODE_CONNECTION_LOST) {
//                reconnectTime = Utils.getRandom(0, 30);
//            }
//            if (BuildConfig.DEBUG) {
//                Log.d("MqttDefaultCallback", "reconnect after " + reconnectTime + " seconds");
//            }
//            ThreadTools.submitTask(() -> {
            MqttAsyncClient mqttAsyncClient = MqttNet.getInstance().getMqttClient();
            if (mqttAsyncClient != null) {
                if (mqttAsyncClient.isConnected()) {
                    MqttNet.getInstance().changeConnectState(PersistentConnectState.CONNECTED);
                    PersistentEventDispatcher.getInstance().handlerMessage(PersistentEventDispatcher.MSG_CONNECT_STATE_CONNECTED);
                } else {
                    if (MqttNet.getInstance().getConnectState() != PersistentConnectState.CONNECTED
                            || MqttNet.getInstance().getConnectState() != PersistentConnectState.CONNECTING) {
                        if (BuildConfig.DEBUG) {
                            Log.d("MqttDefaultCallback", "connectComplete, try reconnect");
                        }
                        try {
                            MqttNet.getInstance().changeConnectState(PersistentConnectState.CONNECTING);
                            mqttAsyncClient.reconnect();
                        } catch (MqttException e) {
                            MqttNet.getInstance().changeConnectState(PersistentConnectState.DISCONNECTED);
                            PersistentEventDispatcher.getInstance().handlerMessage(PersistentEventDispatcher.MSG_CONNECT_STATE_DISCONNECTED);
                            if (BuildConfig.DEBUG) {
                                Log.d("MqttDefaultCallback", "connectComplete, try reconnect fail");
                                e.printStackTrace();
                            }
                        }
                    } else {
                        if (BuildConfig.DEBUG) {
                            Log.d("MqttDefaultCallback", "connectComplete, is connecting");
                        }
                    }
                }
            }
//            }, true, reconnectTime * 1000);
        }
    }

    @Override
    public void connectionLost(Throwable throwable) {
        MqttException mqttException = (MqttException) throwable;
        connectionLostReasonCode = mqttException.getReasonCode();
        MqttNet.getInstance().changeConnectState(PersistentConnectState.DISCONNECTED);
        PersistentEventDispatcher.getInstance().handlerMessage(PersistentEventDispatcher.MSG_CONNECT_STATE_DISCONNECTED);
        if (BuildConfig.DEBUG) {
            Log.d("MqttDefaultCallback", "connectionLost,reason code:" + mqttException.getReasonCode());
            Log.d("MqttDefaultCallback", "connectionLost,cause:" + throwable);
            throwable.printStackTrace();
        }
    }

    @Override
    public void messageArrived(String topic, MqttMessage message) {
        if (BuildConfig.DEBUG) {
            Log.d("MqttDefaultCallback", "messageArrived,topic = [" + topic + "] , msg = [" + new String(message.getPayload(), StandardCharsets.UTF_8) + "],  ");
        }
        try {
            PersistentEventDispatcher.getInstance().handlerMessage(PersistentEventDispatcher.MSG_ARRIVED, topic, message.getPayload());
        } catch (Exception e) {
            if (BuildConfig.DEBUG) {
                Log.d("MqttDefaultCallback", "messageArrived(), send broadcastMsg error");
            }
        }

        if (listenerMap != null && listenerMap.containsKey(topic)) {
            MqttRrpcRequest rpcRequest = new MqttRrpcRequest();
            rpcRequest.setTopic(topic);
            rpcRequest.payloadObj = message.getPayload();
            handleRrpcRequest(rpcRequest, listenerMap.get(topic));
        }
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken deliveryToken) {
        if (BuildConfig.DEBUG) {
            Log.d("MqttDefaultCallback", "deliveryComplete! " + (deliveryToken != null && deliveryToken.getResponse() != null ? deliveryToken.getResponse().getKey() : "null"));
        }
    }

    private void handleRrpcRequest(final MqttRrpcRequest request, final IOnSubscribeRpcListener listener) {
        if (BuildConfig.DEBUG) {
            Log.d("MqttDefaultCallback", "handleRrpcRequest()");
        }
        if (listener != null && listener.getSubscribeListener() != null && request != null) {
            if (listener.getSubscribeListener().needUISafety()) {
                ThreadTools.runOnUiThread(new Runnable() {
                    @Override
                    public void run() {
                        listener.onReceived(request.topic, request, new RrpcResponseHandler(request.topic, listener));
                    }
                });
            } else {
                listener.onReceived(request.topic, request, new RrpcResponseHandler(request.topic, listener));
            }
        }
    }

    public class RrpcResponseHandler implements IOnRrpcResponseHandle {
        private String topic;
        private IOnSubscribeRpcListener listener;

        public RrpcResponseHandler(String topic, IOnSubscribeRpcListener listener) {
            this.topic = topic;
            this.listener = listener;
        }

        @Override
        public void onRrpcResponse(String replyTopic, AResponse response) {
            if (BuildConfig.DEBUG) {
                Log.d("MqttDefaultCallback", "onRrpcResponse(), reply topic = " + replyTopic);
            }
            MqttPublishRequest publisRequest = new MqttPublishRequest();
            publisRequest.isRPC = false;
            if (TextUtils.isEmpty(replyTopic)) {
                publisRequest.topic = topic + "_reply";
            } else {
                publisRequest.topic = replyTopic;
            }

            if (response != null && response.data != null) {
                publisRequest.payloadObj = response.data;
            }

            MqttNet.getInstance().asyncSend(publisRequest, new IOnCallListener() {
                @Override
                public void onSuccess(ARequest request, AResponse response) {
                    if (BuildConfig.DEBUG) {
                        Log.d("MqttDefaultCallback", "onRrpcResponse(), publish succ");
                    }
                    listener.onResponseSuccess(topic);
                }

                @Override
                public void onFailed(ARequest request, AError error) {
                    if (BuildConfig.DEBUG) {
                        Log.d("MqttDefaultCallback", "onRrpcResponse(), publish fail");
                    }
                    listener.onResponseFailed(topic, error);
                }

                @Override
                public boolean needUISafety() {
                    return listener.getSubscribeListener().needUISafety();
                }
            });
        }
    }

}
